package com.aotain.metis.topo;

import org.apache.log4j.Logger;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.kafka.KafkaSpout;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;

import com.aotain.apollo.ApolloConfig;
import com.aotain.apollo.ApolloConfig_Oracle;
import com.aotain.apollo.MetisConfig;
import com.aotain.common.ZkKafkaConfig;
import com.aotain.kafka.KafkaConfigUtil;
import com.aotain.kafka.KafkaProperties;
import com.aotain.metis.bolt.BotsBolt;
import com.aotain.metis.bolt.CCDetectBoltMetis;
import com.aotain.metis.bolt.DomainStatBolt;
import com.aotain.metis.bolt.IPActiveScanBolt;
import com.aotain.metis.bolt.MonAbnReverseStatisBolt;
import com.aotain.metis.bolt.MonSourceAbnormalCalBolt;
import com.aotain.metis.bolt.SQLInjectionBoltMetis;
import com.aotain.metis.bolt.ScanAttackBolt;
import com.aotain.metis.bolt.SourceFmtMonBolt;
import com.aotain.metis.bolt.StatBolt;
import com.aotain.metis.bolt.StatBoltALL;
import com.aotain.metis.bolt.StatBoltMin;
import com.aotain.metis.bolt.ThreatenEvaluateBolt;
import com.aotain.metis.bolt.XSSDetectBoltMetis;


/**
 * 流量监控topo
 * @ClassName: FlowMonitoringTopo 
 * @Description: TODO(这里用一句话描述这个类的作用) 
 * @author 程彬
 * @date 2016年11月30日 下午5:09:58 
 *
 */
public class FlowMonitoringTopo {

	static int workersNum = 6;
	static int spoutNum = 12;
	static int boltNum = 6;
	static int sourcefmtBoltNum = 6;
	public static void main(String[] args) {  

		if (args.length != 5){
			System.err.printf("Usage:<TopoName> <Workers> <SpoutThreadNum> <BoltThreadNum> <SourceFmtboltThreadNum>");
			System.exit(1);
		}  

		workersNum = Integer.parseInt(args[1]);
		spoutNum = Integer.parseInt(args[2]);
		boltNum = Integer.parseInt(args[3]);
		sourcefmtBoltNum = Integer.parseInt(args[4]);

		ZkKafkaConfig z_kConfig = new ZkKafkaConfig("/home/storm/config/dbconfig.ini");
//		Logger.getLogger(FlowMonitoringTopo.class).info("Version 20161130");

		ApolloConfig ap = new ApolloConfig();
		ApolloConfig_Oracle ap_ = new ApolloConfig_Oracle();
		
		MetisConfig metisConf = new MetisConfig();

		String dbJson = ap.getDBObjectJson();
		String dbJson_ = ap_.getDBObjectJson();
		String dbJsonMetis = metisConf.getDBObjectJson();
		Logger.getLogger(FlowMonitoringTopo.class).info(dbJson);
 
		TopologyBuilder builder = new TopologyBuilder();
//		builder.setSpout("sourceSpout", new KafkaSpout2(new KafkaConfigUtil(z_kConfig.getZkConnServer(),KafkaProperties.UDII_Abnormal_Log_topic,KafkaProperties.zkRoot_FlowMon,KafkaProperties.Log_UDII_FlowMon_groupId ).getConfig()),spoutNum);
		builder.setSpout("sourceSpout", new KafkaSpout(new KafkaConfigUtil(z_kConfig.getZkConnServer(),KafkaProperties.UDII_Abnormal_Log_topic,KafkaProperties.zkRoot_FlowMon,KafkaProperties.Log_UDII_FlowMon_groupId ).getConfig()),spoutNum);
		builder.setBolt("SourceFmtBolt", new SourceFmtMonBolt(),sourcefmtBoltNum).shuffleGrouping("sourceSpout");
		builder.setBolt("StatBoltMin", new StatBoltMin(),boltNum).fieldsGrouping("SourceFmtBolt", new Fields("sip"));
		builder.setBolt("DomainStatBolt", new DomainStatBolt(),boltNum).fieldsGrouping("SourceFmtBolt", new Fields("sip"));
		
		builder.setBolt("StatBolt", new StatBolt(),boltNum).fieldsGrouping("StatBoltMin", new Fields("sip"));
		builder.setBolt("StatBoltALL", new StatBoltALL(),boltNum).fieldsGrouping("StatBoltMin", new Fields("dateStr30"));
		
		builder.setBolt("CCDetectBoltMetis", new CCDetectBoltMetis(),boltNum).fieldsGrouping("SourceFmtBolt", new Fields("sip"));
		builder.setBolt("XSSDetectBoltMetis", new XSSDetectBoltMetis(),boltNum).fieldsGrouping("SourceFmtBolt", new Fields("sip"));
		builder.setBolt("AbnormalCalBoltMetis", new MonSourceAbnormalCalBolt(),boltNum).fieldsGrouping("SourceFmtBolt", new Fields("sip"));
		builder.setBolt("SQLInjectionBoltMetis", new SQLInjectionBoltMetis(),boltNum).fieldsGrouping("SourceFmtBolt", new Fields("sip"));
		builder.setBolt("ScanAttackBolt", new ScanAttackBolt(),boltNum).fieldsGrouping("SourceFmtBolt", new Fields("sip"));
		builder.setBolt("BotsBolt", new BotsBolt(),boltNum).fieldsGrouping("SourceFmtBolt", new Fields("sip"));
		
		builder.setBolt("MonAbnReverseStatisBolt", new MonAbnReverseStatisBolt(),boltNum)
			.fieldsGrouping("AbnormalCalBoltMetis", new Fields("sip"))
			.fieldsGrouping("CCDetectBoltMetis", new Fields("sip"))
			.fieldsGrouping("XSSDetectBoltMetis", new Fields("sip"))
			.fieldsGrouping("SQLInjectionBoltMetis", new Fields("sip"))
			.fieldsGrouping("ScanAttackBolt", new Fields("sip"))
			.fieldsGrouping("BotsBolt", new Fields("sip"));
		
		builder.setBolt("ThreatenEvaluateBolt", new ThreatenEvaluateBolt(),boltNum)
		.fieldsGrouping("AbnormalCalBoltMetis", new Fields("sip"))
		.fieldsGrouping("CCDetectBoltMetis", new Fields("sip"))
		.fieldsGrouping("XSSDetectBoltMetis", new Fields("sip"))
		.fieldsGrouping("SQLInjectionBoltMetis", new Fields("sip"))
		.fieldsGrouping("ScanAttackBolt", new Fields("sip"))
		.fieldsGrouping("BotsBolt", new Fields("sip"));
		
		builder.setBolt("IPActiveScanBolt", new IPActiveScanBolt(),boltNum).fieldsGrouping("StatBoltMin", new Fields("sip"));
		
		
		Config conf = new Config();    
		conf.setNumWorkers(workersNum); 
		conf.put("mushroomserver", z_kConfig.getMushroomServer());
		conf.put("brokerlist", z_kConfig.getBrokerList());
		conf.put("dbjson", dbJson);
		conf.put("dbjson_", dbJson_);
		conf.put("dbJsonMetis", dbJsonMetis);
		conf.put("sqlinjection", z_kConfig.getSQLInjection());
		Logger.getLogger(FlowMonitoringTopo.class).info(dbJson);
		conf.put("zkConn", z_kConfig.getZkConnServer());
		//		conf.put("topology.acker.executors", 10);
		conf.setNumAckers(0);
		//		conf.put("storm.messaging.netty.max_wait_ms", 10000);
		conf.setDebug(false);

		if (args.length > 0) {  
			try {
				StormSubmitter.submitTopology(args[0], conf, builder.createTopology());

			} catch (AlreadyAliveException e) {
				e.printStackTrace();
			} catch (InvalidTopologyException e) {
				e.printStackTrace();
			}	catch (AuthorizationException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}else {
			LocalCluster localCluster = new LocalCluster();
			localCluster.submitTopology("FlowMonitoringTopo", conf, builder.createTopology());
		}
	}

}
